package com.atguigu.flink.chapter10;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/6/27 09:23
 */
public class FileConnectorDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        // 通过 ddl 的方式创建连接器表
        tEnv.executeSql("create table sensor(" +
                            " id string, " +
                            " ts bigint, " +
                            " vc int " +
                            ")with(" +
                            " 'connector' = 'filesystem', " +
                            " 'path' = 'input/sensor.json', " +
                            " 'format' = 'json' " +
                            ")");
    
        Table result = tEnv.sqlQuery("select * from sensor where id='sensor_1'");
        tEnv.executeSql("create table s1(" +
                            " a string, " +
                            " b bigint, " +
                            " d int " +
                            ")with(" +
                            " 'connector' = 'filesystem', " +
                            " 'path' = 'input/s1.txt', " +
                            " 'format' = 'json' " +
                            ")");
        
        // insert into ... select ....
//        tEnv.executeSql("insert into s1 select id, ts, vc from " + result);
        result.executeInsert("s1");
    
    
    }
}
